with f as (
  select
        id                       
       ,appeal_no                
       ,waybill_no               
       ,user_name                
       ,phone                    
       ,status                   
       ,type                     
       ,cause                    
       ,receiver_address         
       ,complaint_time           
       ,province_id              
       ,province_desc            
       ,city_id                  
       ,city_desc                
       ,description              
       ,accept_network_id        
       ,accept_network_code      
       ,accept_network_name      
       ,accept_time              
       ,assign_time              
       ,assign_type              
       ,accept_user_id           
       ,accept_user_code         
       ,accept_user_name         
       ,is_upgrade               
       ,upgrade_network_id       
       ,upgrade_network_code     
       ,upgrade_network_name     
       ,upgrade_time             
       ,process_user_id          
       ,process_user_code        
       ,process_user_name        
       ,process_time             
       ,create_time              
       ,update_time              
       ,accept_network_type      
       ,accept_proxy_id          
       ,accept_proxy_code        
       ,accept_proxy_name        
       ,accept_franchisee_id     
       ,accept_franchisee_code   
       ,accept_franchisee_name   
       ,complain_type            
       ,is_auto_allocation       
       ,dealwith                 
       ,sender_name              
       ,sender_phone             
       ,sender_address           
       ,receiver_name            
       ,receiver_phone
,accept_network_mr_code
,accept_network_mr_Name
,transfer_order_time
,resp_status
,delivery_code
,delivery_name
,is_dispatcher
,receiver_province_name
,receiver_province_id
,receiver_city_name
,receiver_city_id
,receiver_area_name
,receiver_area_id
,product_type
       ,row_number() over(partition by waybill_no order by update_time desc) as rank
  from jms_ods.offline_complaint
    where dt between date_sub('{{ execution_date | cst_ds }}',70) and '{{ execution_date | cst_ds }}'
             and date(create_time) between date_sub('{{ execution_date | cst_ds }}',60)
             and '{{ execution_date | cst_ds }}'
)
insert overwrite table jms_dwd.dwd_offline_complaint_dt partition(dt)
  select
        id                       
       ,appeal_no                
       ,waybill_no               
       ,user_name                
       ,phone                    
       ,status                   
       ,type                     
       ,cause                    
       ,receiver_address         
       ,complaint_time           
       ,province_id              
       ,province_desc            
       ,city_id                  
       ,city_desc                
       ,description              
       ,accept_network_id        
       ,accept_network_code      
       ,accept_network_name      
       ,accept_time              
       ,assign_time              
       ,assign_type              
       ,accept_user_id           
       ,accept_user_code         
       ,accept_user_name         
       ,is_upgrade               
       ,upgrade_network_id       
       ,upgrade_network_code     
       ,upgrade_network_name     
       ,upgrade_time             
       ,process_user_id          
       ,process_user_code        
       ,process_user_name        
       ,process_time             
       ,create_time              
       ,update_time              
       ,accept_network_type      
       ,accept_proxy_id          
       ,accept_proxy_code        
       ,accept_proxy_name        
       ,accept_franchisee_id     
       ,accept_franchisee_code   
       ,accept_franchisee_name   
       ,complain_type            
       ,is_auto_allocation       
       ,dealwith                 
       ,sender_name              
       ,sender_phone             
       ,sender_address           
       ,receiver_name            
       ,receiver_phone
,accept_network_mr_code
,accept_network_mr_Name
,transfer_order_time
,resp_status
,delivery_code
,delivery_name
,is_dispatcher
,receiver_province_name
,receiver_province_id
,receiver_city_name
,receiver_city_id
,receiver_area_name
,receiver_area_id
,product_type
       ,date(create_time) as dt
 from f
   where rank = 1
          distribute by dt ;
